package com.cdy.client.handler;

import com.cdy.common.response.*;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 客户端响应处理器
 *
 * Created by 陈东一
 * 2019/10/20 0020 12:36
 */
@Slf4j
public class HandlerManager implements ResponseHandler {

    private final HashMap<String, ResponseHandler> map = new HashMap<>();

    private int highReadWaterMarker = 900, lowReadWaterMarker = 600;
    private ThreadPoolExecutor executor =
            new ThreadPoolExecutor(8, 16, 1, TimeUnit.MINUTES,
                    new ArrayBlockingQueue<Runnable>(1000), new ThreadPoolExecutor.DiscardOldestPolicy());

    public HandlerManager() {
        map.put(RequestUploadResponse.class.getName(), new RequestUploadResponseHandler());
        map.put(UploadResponse.class.getName(), new UploadResponseHandler());
        map.put(RequestDownloadResponse.class.getName(), new RequestDownloadResponseHandler());
        map.put(DownloadResponse.class.getName(), new DownloadResponseHandler());
    }


    @Override
    public void handle(ChannelHandlerContext ctx, Response msg)  {
        int queued = executor.getQueue().size();
        if (queued > highReadWaterMarker) {
            ctx.channel().config().setAutoRead(false);
        }
        if (queued < lowReadWaterMarker) {
            ctx.channel().config().setAutoRead(true);
        }

        executor.execute(()->{
            ResponseHandler commandHandler = map.get(msg.getClass().getName());
            if (commandHandler!=null) {
                commandHandler.handle(ctx, msg);
            } else {
                log.warn("no such ResponseHandler {} ",msg);
            }
        });

    }
}
